package com.example.buffer;


import org.junit.jupiter.api.Test;

import java.io.*;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;

/**
 * @author 86130-leon
 * @date 2024/1/10--15:44
 * 词频统计
 */
public class WordCount {
    final ForkJoinPool pool = ForkJoinPool.commonPool();//是一个线程池
    /**
     * 统计词频   word是自己生成的大文件，固定单词长度是5
     * 单个线程统计词频;  每次缓冲区可能截取的单词不是完整的
     * @throws IOException
     */
    @Test
    public void compare_with_single() throws IOException {
        var in = new BufferedInputStream(new FileInputStream("word"));//读取一个大文件
        var buf = new byte[4 * 1024];
        var len = 0;
        var total = new HashMap<String, Integer>();
        var startTime = System.currentTimeMillis();
        while ((len = in.read(buf)) != -1) {
            var bytes = Arrays.copyOfRange(buf, 0, len);
            var str = new String(bytes);
            var hashMap = countByString(str);
            for (var entry : hashMap.entrySet()) {
                var key = entry.getKey();
                incKey(key, total, entry.getValue());// 累加每次缓冲区的词频统计
            }
        }
        System.out.println("time: " + (System.currentTimeMillis() - startTime) + "ms");
        System.out.println(total.get("ababb"));
        System.out.println(total.size());

    }

    /**
     * 并发执行: 容易出现IO/OOM/...异常
     * 使用 ForkJoinPool 线程池并发的执行task 进行词频统计;  有误差-->每个线程获取的文件区域不一定是完整的单词
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @Test
    public void count() throws ExecutionException, InterruptedException {
        var counter=new WordCount();
        System.out.println("当前机器的核数: "+Runtime.getRuntime().availableProcessors());
        counter.run("word",1024);
    }


    /**
     * 并发任务
     */
    class CountTask implements Callable<HashMap<String, Integer>> {
        private final long start;
        private final long end;
        private final String fileName;

        public CountTask(String fileName, long start, long end) {
            this.start = start;
            this.end = end;
            this.fileName = fileName;
        }

        @Override
        public HashMap<String, Integer> call() throws Exception {
            var channel = new RandomAccessFile(this.fileName, "rw").getChannel();//可以从文件的任意位置读取文件
            // [start,end] -> Memory
            // Device -> Kernel Space -> UserSpace(buffer) -> Thread
            var mbuf = channel.map(FileChannel.MapMode.READ_ONLY, this.start, this.end); //相当于拿到文件的一段对应到内存区域
            var str = StandardCharsets.US_ASCII.decode(mbuf).toString();//用指定的编码解码; 因为都是单词所以ASCII解码也可以
            return countByString(str);
        }
    }

    public void run(String fileName, long chunkSize) throws ExecutionException, InterruptedException {
        var file = new File(fileName);
        var fileSize = file.length();
        long position = 0;
        var start = System.currentTimeMillis();
        var tasks = new ArrayList<Future<HashMap<String, Integer>>>();
        while (position < fileSize) {
            var next = Math.min(position + chunkSize, fileSize);
            var task = new CountTask(fileName, position, next);
            position = next;
            var future = pool.submit(task);//提交任务到线程池
            tasks.add(future);
        }
        System.out.format("split to %d tasks\n",tasks.size());

        //做一次同步
        var totalMap = new HashMap<String, Integer>();
        for (var future : tasks) {
            var map = future.get();
            for (var entry : map.entrySet()) {
                incKey(entry.getKey(), totalMap, entry.getValue());
            }
        }
        System.out.println("time = " + (System.currentTimeMillis() - start) + "ms");
        System.out.println("total = "+totalMap.size());
        System.out.println(totalMap.get("ababb"));//打印统计信息

    }

    private static HashMap<String, Integer> countByString(String str) {
        var map = new HashMap<String, Integer>();
        StringTokenizer tokenizer = new StringTokenizer(str);
        while (tokenizer.hasMoreTokens()) {
            var word = tokenizer.nextToken();
            incKey(word, map, 1);
        }
        return map;
    }

    private static void incKey(String key, HashMap<String, Integer> map, Integer n) {
        if (map.containsKey(key)) {
            map.put(key, map.get(key) + n);
        } else {
            map.put(key, n);
        }

    }



}
